package com.shujia.spark

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo12Cache {
  def main(args: Array[String]): Unit = {

    /**
      * 缓存
      *
      * 如果不适用缓存 同一个rdd被多次使用 会计算多次
      *
      */


    val conf = new SparkConf().setAppName("Demo11Student").setMaster("local[4]")
    val sc = new SparkContext(conf)

    //自动checkpoint路径
    //如果是本地模式  就是本地路径
    //如果是集群模式运行，就是hdfs路径
    sc.setCheckpointDir("spark/data/checkpoint")


    val rdd = sc.textFile("spark/data/student.txt")


    var tupleRDD = rdd.map(line => {

      println("=========")

      val split = line.split(",")
      val id = split(0)
      val clazz = split(4)
      val gender = split(3)
      (id, clazz, gender)
    })

    //对重复使用的rdd进行缓存  cache  将数据缓存到内存中
    //tupleRDD = tupleRDD.cache()
    //tupleRDD = tupleRDD.persist(StorageLevel.MEMORY_ONLY)

    //DISK_ONLY  将数据缓存到磁盘
    //persist  懒执行 需要action算子触发

    /**
      * 缓存级别选择
      * 1、MEMORY_ONLY  内存充足
      * 2、MEMORY_ONLY_SER   压缩放内存
      *
      *
      * 不管是否压缩放内存都比放磁盘块
      * MEMORY_AND_DISK
      * MEMORY_AND_DISK_SER
      * 3、MEMORY_AND_DISK_SER   压缩只有放内存  内存放不下再 放此案
      */
    tupleRDD = tupleRDD.persist(StorageLevel.DISK_ONLY)


    //将数据写入hdfs
    /**
      * sc.setCheckpointDir("spark/data/checkpoint")
      *
      * checkpoint  当第一个job执行完成之后，会向前回溯
      * 如果发现某一个rdd执行了checkpoint  那么会重新启动一个job计算这个rdd的数据，
      * 将这个rdd的数据写入hdfs
      *
      * checkpoint 可以切断rdd依赖关系   后面的rdd不在依赖前面rdd，而是依赖hdfs里面的数据
      *
      * 在checkpoint  之前先对rdd进行一次cache
      *
      */
    tupleRDD.checkpoint()


    tupleRDD
      .map(kv => (kv._2, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)


    tupleRDD
      .map(kv => (kv._3, 1))
      .reduceByKey((x, y) => x + y)
      .foreach(println)

  }
}
